#include "util/fileaio/aio_file_reader.h"

namespace hawking {
namespace indexlib {

DEFINE_int32(io_uring_queue_size, 128, "default io_uring_queue_size");

DiskFileReadProcesser::~DiskFileReadProcesser() {
    start_.store(false, std::memory_order_relaxed);

    // TODO optim
    io_uring_sqe* sqe = io_uring_get_sqe(&ring_);
    io_uring_prep_nop(sqe);
    io_uring_submit(&ring_);
    if (io_uring_dispose_thread_.joinable()) {
        io_uring_dispose_thread_.join();
    }
    if (stat_thread_.joinable()) {
        stat_thread_.join();
    }

    io_uring_queue_exit(&ring_);
    close(fd_);
}

bool DiskFileReadProcesser::Init(
    const AioOption& opt, const char* file_path, bool need_stat_thread) {
    int32_t flags = O_RDONLY | O_NONBLOCK;
    fd_ = open(file_path, flags);
    if (fd_ == -1) {
        PLOG(ERROR) << "open file fail " << file_path;
        return false;
    }
    filename_ = file_path;

    // use default mode
    int32_t init_ret = io_uring_queue_init(opt.io_uring_queue_size, &ring_, 0);
    if (init_ret) {
        LOG(ERROR) << "io_uring_queue_init fail with queue size "
           << opt.io_uring_queue_size << ", errno " << (-init_ret);
        return false;
    }

    start_.store(true, std::memory_order_relaxed);
    io_uring_dispose_thread_ = std::thread([this](){
        struct io_uring* ring = &this->ring_;
        io_uring_cqe* dummy = nullptr;
        //__kernel_timespec ts{.tv_sec = 1, .tv_nsec = 0};
        while (1) {
            if (!this->start_.load(std::memory_order_relaxed)) {
                io_uring_peek_cqe(ring, &dummy);
                if (!dummy) {
                    LOG(INFO) << "complete queue empty. break.";
                    break;
                }
            }

            io_uring_cqe* cqe = nullptr;
            // TODO io_uring_wait_cqe_timeout would lead large bad perf
            //int32_t ret = io_uring_wait_cqe_timeout(ring, &cqe, &ts);
            int32_t ret = io_uring_wait_cqe(ring, &cqe);
            if (ret < 0 || !cqe) {
                LOG(ERROR) << "io_uring_wait_cqe ret " << (-ret);
                continue;
            }
            // TODO optim
            if (!this->start_.load(std::memory_order_relaxed)) {
                break;
            }

            DiskFileArgs* arg = reinterpret_cast<DiskFileArgs*>(cqe->user_data);
            if (likely(cqe != nullptr)) {
                arg->size = cqe->res;
                if (unlikely(cqe->res < 0)) {
                    LOG(ERROR) << "cqe->res " << cqe->res;
                    arg->status = Status::IOError;
                }
            }
            arg->cv.notify_one();
            arg->done_flag.store(1, std::memory_order_relaxed);
            io_uring_cqe_seen(ring, cqe);
            this->read_response_count_.fetch_add(1, std::memory_order_relaxed);
        }
    });

    if (need_stat_thread) {
        stat_thread_ = std::thread([this](){
            while (start_.load(std::memory_order_relaxed)) {
                uint64_t request_sum = this->read_request_count_.load(std::memory_order_relaxed);
                uint64_t response_sum = this->read_response_count_.load(std::memory_order_relaxed);
                LOG(INFO) << "file[" << this->filename_ << "], request num " << request_sum
                    << ", response num " << response_sum
                    << ", concurrnecy " << this->concurrency_.load(std::memory_order_relaxed);
                // TODO(mander.liu) bvar update
                std::this_thread::sleep_for(std::chrono::seconds(1));
            }
        });
    }

    return true;
}

bool DiskFileReadProcesser::Read(off_t offset, DiskFileArgs* arg) {
    while (1) {
        int32_t concurrency = concurrency_.load(std::memory_order_relaxed);
        if (concurrency < max_concurrency_ &&
            concurrency_.compare_exchange_strong(concurrency, concurrency + 1)) {
            struct io_uring_sqe* sqe = io_uring_get_sqe(&ring_);
            if (!sqe) {
                LOG(ERROR) << "io_uring_get_sqe fail: errno : ";
                concurrency_.fetch_sub(1, std::memory_order_relaxed);
                return false;
            }
            io_uring_prep_readv(sqe, fd_, &arg->iov, 1, offset);
            io_uring_sqe_set_data(sqe, arg);
            io_uring_submit(&ring_);
            int32_t ret = io_uring_submit(&ring_);
            if (ret < 0) {
                LOG(ERROR) << "io_uring_submit fail: errno : " << ret;
                concurrency_.fetch_sub(1, std::memory_order_relaxed);
                return false;
            }
            read_request_count_.fetch_add(1, std::memory_order_relaxed);
            break;
        }

        bthread_usleep(10);
    }

    return true;
    // call Wait outside
}

ReturnValue<int32_t> DiskFileReadProcesser::Read(uint8_t* buffer, size_t length, off_t offset) {
    std::unique_ptr<DiskFileArgs> arg = std::make_unique<DiskFileArgs>(buffer, length);
    while (1) {
        int32_t concurrency = concurrency_.load(std::memory_order_relaxed);
        if (concurrency < max_concurrency_ &&
            concurrency_.compare_exchange_strong(concurrency, concurrency + 1)) {
            struct io_uring_sqe* sqe = io_uring_get_sqe(&ring_);
            if (!sqe) {
                LOG(ERROR) << "io_uring_get_sqe fail: errno : ";
                concurrency_.fetch_sub(1, std::memory_order_relaxed);
                return ReturnValue<int32_t>{Status::IOError, -1};
            }
            io_uring_prep_readv(sqe, fd_, &arg->iov, 1, offset);
            io_uring_sqe_set_data(sqe, arg.get());
            io_uring_submit(&ring_);
            int32_t ret = io_uring_submit(&ring_);
            if (ret < 0) {
                LOG(ERROR) << "io_uring_submit fail: errno : " << ret;
                concurrency_.fetch_sub(1, std::memory_order_relaxed);
                return ReturnValue<int32_t>{Status::IOError, -1};
            }
            read_request_count_.fetch_add(1, std::memory_order_relaxed);
            break;
        }

        bthread_usleep(10);
    }

    Wait(*arg);
    return ReturnValue<int32_t>{arg->status, arg->size};
}

void DiskFileReadProcesser::Wait(DiskFileArgs& arg) {
    bthread::Mutex mutex;
    std::unique_lock<bthread::Mutex> lock(mutex);
    while (!arg.done_flag.load(std::memory_order_relaxed)) {
        arg.cv.wait_for(lock, 10);
    }
    concurrency_.fetch_sub(1, std::memory_order_relaxed);
}

}
}
